#include <srs_app_http_hooks.hpp>

#ifdef SRS_AUTO_HTTP_CALLBACK

#include <sstream>
using namespace std;

#include <srs_kernel_error.hpp>
#include <srs_rtmp_stack.hpp>
#include <srs_app_st.hpp>
#include <srs_protocol_json.hpp>
#include <srs_app_dvr.hpp>
#include <srs_app_http_client.hpp>
#include <srs_core_autofree.hpp>
#include <srs_app_config.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_app_http_conn.hpp>
#include <srs_app_utility.hpp>

#define SRS_HTTP_RESPONSE_OK    SRS_XSTR(ERROR_SUCCESS)

#define SRS_HTTP_HEADER_BUFFER        1024
#define SRS_HTTP_READ_BUFFER    4096
#define SRS_HTTP_BODY_BUFFER        32 * 1024

// the timeout for hls notify, in us.
#define SRS_HLS_NOTIFY_TIMEOUT_US (int64_t)(10*1000*1000LL)

SrsHttpHooks::SrsHttpHooks() {
}

SrsHttpHooks::~SrsHttpHooks() {
}

int SrsHttpHooks::on_connect(const string& url, SrsRequest* req) {
    int ret = ERROR_SUCCESS;
    int client_id = _srs_context->get_id();

    stringstream ss;
    ss << SRS_JOBJECT_START
       << SRS_JFIELD_STR("action", "on_connect") << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("client_id", client_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("ip", req->ip) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("vhost", req->vhost) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("app", req->app) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("tcUrl", req->tcUrl) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("pageUrl", req->pageUrl)
       << SRS_JOBJECT_END;

    string data = ss.str();
    string res;
    int status_code;
    if ((ret = do_post(url, data, status_code, res)) != ERROR_SUCCESS) {
        srs_error("http post on_connect uri failed. "
                  "client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
                  client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
        return ret;
    }

    srs_trace("http hook on_connect success. client_id=%d, url=%s, request=%s, response=%s, ret=%d",
              client_id, url.c_str(), data.c_str(), res.c_str(), ret);
    return ret;
}

void SrsHttpHooks::on_close(const string& url, SrsRequest* req, int64_t send_bytes, int64_t recv_bytes) {
    int ret = ERROR_SUCCESS;
    int client_id = _srs_context->get_id();

    stringstream ss;
    ss << SRS_JOBJECT_START
       << SRS_JFIELD_STR("action", "on_close") << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("client_id", client_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("ip", req->ip) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("vhost", req->vhost) << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("send_bytes", send_bytes) << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("recv_bytes", recv_bytes) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("app", req->app)
       << SRS_JOBJECT_END;

    string data = ss.str();
    string res;
    int status_code;
    if ((ret = do_post(url, data, status_code, res)) != ERROR_SUCCESS) {
        srs_warn("http post on_close uri failed, ignored. "
                 "client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
                 client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
        return;
    }

    srs_trace("http hook on_close success. client_id=%d, url=%s, request=%s, response=%s, ret=%d",
              client_id, url.c_str(), data.c_str(), res.c_str(), ret);
    return;
}

int SrsHttpHooks::on_publish(const string& url, SrsRequest* req, string& resp) {
    int ret = ERROR_SUCCESS;
    int client_id = _srs_context->get_id();

    stringstream ss;
    ss << SRS_JOBJECT_START
       << SRS_JFIELD_STR("action", "on_publish") << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("client_id", client_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("ip", req->ip) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("vhost", req->vhost) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("app", req->app) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("tcUrl", req->tcUrl) << SRS_JFIELD_CONT  // Add tcUrl for auth publish rtmp stream client
       << SRS_JFIELD_STR("stream", req->stream) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream_id", req->stream_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("param", req->param)
       << SRS_JOBJECT_END;

    resp.clear();
    string data = ss.str();
    int status_code;
    if ((ret = do_post(url, data, status_code, resp)) != ERROR_SUCCESS) {
        srs_error("http post on_publish uri failed. client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
                  client_id, url.c_str(), data.c_str(), resp.c_str(), status_code, ret);
        return ret;
    }

    srs_trace("http hook on_publish success. client_id=%d, url=%s, request=%s, response=%s, ret=%d",
              client_id, url.c_str(), data.c_str(), resp.c_str(), ret);
    return ret;
}

void SrsHttpHooks::on_unpublish(const string& url, SrsRequest* req) {
    int ret = ERROR_SUCCESS;
    int client_id = _srs_context->get_id();

    stringstream ss;
    ss << SRS_JOBJECT_START
       << SRS_JFIELD_STR("action", "on_unpublish") << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("client_id", client_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("ip", req->ip) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("vhost", req->vhost) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("app", req->app) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream", req->stream) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream_id", req->stream_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("param", req->param)
       << SRS_JOBJECT_END;

    string data = ss.str();
    string res;
    int status_code;
    if ((ret = do_post(url, data, status_code, res)) != ERROR_SUCCESS) {
        srs_warn("http post on_unpublish uri failed, ignored. "
                 "client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
                 client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
        return;
    }

    srs_trace("http hook on_unpublish success. client_id=%d, url=%s, request=%s, response=%s, ret=%d",
              client_id, url.c_str(), data.c_str(), res.c_str(), ret);
    return;
}

int SrsHttpHooks::on_play(const string& url, SrsRequest* req) {
    int ret = ERROR_SUCCESS;
    int client_id = _srs_context->get_id();

    stringstream ss;
    ss << SRS_JOBJECT_START
       << SRS_JFIELD_STR("action", "on_play") << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("client_id", client_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("ip", req->ip) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("vhost", req->vhost) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("app", req->app) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream", req->stream) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream_id", req->stream_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("param", req->param) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("pageUrl", req->pageUrl)
       << SRS_JOBJECT_END;

    string data = ss.str();
    string res;
    int status_code;
    if ((ret = do_post(url, data, status_code, res)) != ERROR_SUCCESS) {
        srs_error("http post on_play uri failed. "
                  "client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
                  client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
        return ret;
    }

    srs_trace("http hook on_play success. client_id=%d, url=%s, request=%s, response=%s, ret=%d",
              client_id, url.c_str(), data.c_str(), res.c_str(), ret);

    return ret;
}

void SrsHttpHooks::on_stop(const string& url, SrsRequest* req) {
    int ret = ERROR_SUCCESS;
    int client_id = _srs_context->get_id();

    stringstream ss;
    ss << SRS_JOBJECT_START
       << SRS_JFIELD_STR("action", "on_stop") << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("client_id", client_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("ip", req->ip) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("vhost", req->vhost) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("app", req->app) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream", req->stream) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream_id", req->stream_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("param", req->param)
       << SRS_JOBJECT_END;

    string data = ss.str();
    string res;
    int status_code;
    if ((ret = do_post(url, data, status_code, res)) != ERROR_SUCCESS) {
        srs_warn("http post on_stop uri failed, ignored. "
                 "client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
                 client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
        return;
    }

    srs_trace("http hook on_stop success. client_id=%d, url=%s, request=%s, response=%s, ret=%d",
              client_id, url.c_str(), data.c_str(), res.c_str(), ret);
    return;
}

int SrsHttpHooks::on_dvr(int cid, const string& url, SrsRequest* req, const string& file) {
    int ret = ERROR_SUCCESS;
    int client_id = cid;
    auto& cwd = _srs_config->cwd();

    stringstream ss;
    ss << SRS_JOBJECT_START
       << SRS_JFIELD_STR("action", "on_dvr") << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("client_id", client_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("ip", req->ip) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("vhost", req->vhost) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("app", req->app) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream", req->stream) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream_id", req->stream_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("param", req->param) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("cwd", cwd) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("file", file)
       << SRS_JOBJECT_END;

    string data = ss.str();
    string res;
    int status_code;
    if ((ret = do_post(url, data, status_code, res)) != ERROR_SUCCESS) {
        srs_error("http post on_dvr uri failed, ignored. "
                  "client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
                  client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
        return ret;
    }

    srs_trace("http hook on_dvr success. "
              "client_id=%d, url=%s, request=%s, response=%s, ret=%d",
              client_id, url.c_str(), data.c_str(), res.c_str(), ret);
    return ret;
}

int SrsHttpHooks::on_hls(int cid, const string& url, SrsRequest* req, const string& file, string& ts_url,
                         const string& m3u8, const string& m3u8_url, int sn, double duration) {
    int ret = ERROR_SUCCESS;
    int client_id = cid;
    auto& cwd = _srs_config->cwd();

    // the ts_url is under the same dir of m3u8_url.
    string prefix = srs_path_dirname(m3u8_url);
    if (!prefix.empty() && !srs_string_is_http(ts_url)) {
        ts_url = prefix + "/" + ts_url;
    }

    stringstream ss;
    ss << SRS_JOBJECT_START
       << SRS_JFIELD_STR("action", "on_hls") << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("client_id", client_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("ip", req->ip) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("vhost", req->vhost) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("app", req->app) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream", req->stream) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("stream_id", req->stream_id) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("param", req->param) << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("duration", duration) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("cwd", cwd) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("file", file) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("url", ts_url) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("m3u8", m3u8) << SRS_JFIELD_CONT
       << SRS_JFIELD_STR("m3u8_url", m3u8_url) << SRS_JFIELD_CONT
       << SRS_JFIELD_ORG("seq_no", sn)
       << SRS_JOBJECT_END;

    string data = ss.str();
    string res;
    int status_code;
    if ((ret = do_post(url, data, status_code, res)) != ERROR_SUCCESS) {
        srs_error("http post on_hls uri failed, ignored. "
                  "client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
                  client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
        return ret;
    }

    srs_trace("http hook on_hls success. "
              "client_id=%d, url=%s, request=%s, response=%s, ret=%d",
              client_id, url.c_str(), data.c_str(), res.c_str(), ret);
    return ret;
}

int SrsHttpHooks::on_hls_notify(int cid, string& url, SrsRequest* req, const string& ts_url, int nb_notify) {
    int ret = ERROR_SUCCESS;
    int client_id = cid;

    if (srs_string_is_http(ts_url)) {
        url = ts_url;
    }
    string nurl("action=on_hls_notify");

    if (srs_string_breplace(url, "[app]", req->app) == false) {
        nurl += "&app=" + req->app;
    }
    if (srs_string_breplace(url, "[stream]", req->stream) == false) {
        nurl += "&stream=" + req->stream;
    }
    if (srs_string_breplace(url, "[stream_id]", req->stream_id) == false) {
        nurl += "&stream_id=" + req->stream_id;
    }
    if (srs_string_breplace(url, "[ts_url]", ts_url) == false) {
        nurl += "&ts_url=" + ts_url;
    }
    if (srs_string_breplace(url, "[param]", req->param) == false) {
        nurl += "&param=" + req->param;
    }

    if (url.find('?') == string::npos) {
        url += "?" + nurl;
    } else {
        url += "&" + nurl;
    }

    int64_t starttime = srs_update_system_time_ms();

    SrsHttpUri uri;
    if ((ret = uri.initialize(url)) != ERROR_SUCCESS) {
        srs_error("http: post failed. url=%s, ret=%d", url.c_str(), ret);
        return ret;
    }

    SrsHttpClient http;
    if ((ret = http.initialize(uri.get_host(), uri.get_port(), SRS_HLS_NOTIFY_TIMEOUT_US)) != ERROR_SUCCESS) {
        return ret;
    }

    string path = uri.get_query();
    if (path.empty()) {
        path = uri.get_path();
    } else {
        path = uri.get_path();
        path += "?";
        path += uri.get_query();
    }
    srs_warn("GET %s", path.c_str());

    ISrsHttpMessage* msg = NULL;
    if ((ret = http.get(path.c_str(), "", &msg)) != ERROR_SUCCESS) {
        return ret;
    }
    SrsAutoFree(ISrsHttpMessage, msg);

    int nb_buf = srs_min(nb_notify, SRS_HTTP_READ_BUFFER);
    char* buf = new char[nb_buf];
    SrsAutoFreeA(char, buf);

    int nb_read = 0;
    ISrsHttpResponseReader* br = msg->body_reader();
    while (nb_read < nb_notify && !br->eof()) {
        int nb_bytes = 0;
        if ((ret = br->read(buf, nb_buf, &nb_bytes)) != ERROR_SUCCESS) {
            break;
        }
        nb_read += nb_bytes;
    }

    int spenttime = (int)(srs_update_system_time_ms() - starttime);
    srs_trace("http hook on_hls_notify success. client_id=%d, url=%s, code=%d, spent=%dms, read=%dB, ret=%d",
              client_id, url.c_str(), msg->status_code(), spenttime, nb_read, ret);

    // ignore any error for on_hls_notify.
    ret = ERROR_SUCCESS;
    return ret;
}

int SrsHttpHooks::do_post(const string& url, const string& req, int& code, string& res) {
    int ret = ERROR_SUCCESS;

    SrsHttpUri uri;
    if ((ret = uri.initialize(url)) != ERROR_SUCCESS) {
        srs_error("http: post failed. url=%s, ret=%d", url.c_str(), ret);
        return ret;
    }

    SrsHttpClient http;
    if ((ret = http.initialize(uri.get_host(), uri.get_port())) != ERROR_SUCCESS) {
        return ret;
    }

    ISrsHttpMessage* msg = NULL;
    if ((ret = http.post(uri.get_path(), req, &msg)) != ERROR_SUCCESS) {
        return ret;
    }
    SrsAutoFree(ISrsHttpMessage, msg);

    code = msg->status_code();
    if ((ret = msg->body_read_all(res)) != ERROR_SUCCESS) {
        return ret;
    }

    // ensure the http status is ok.
    if (code != SRS_CONSTS_HTTP_OK) {
        ret = ERROR_HTTP_STATUS_INVALID;
        srs_error("invalid response status=%d. ret=%d", code, ret);
        return ret;
    }

    // should never be empty.
    if (res.empty()) {
        ret = ERROR_HTTP_DATA_INVALID;
        srs_error("invalid empty response. ret=%d", ret);
        return ret;
    }

    // parse string res to json.
    SrsJsonAny* info = SrsJsonAny::loads((char*)res.c_str());
    if (!info) {
        ret = ERROR_HTTP_DATA_INVALID;
        srs_error("invalid response %s. ret=%d", res.c_str(), ret);
        return ret;
    }
    SrsAutoFree(SrsJsonAny, info);

    // response error code in string.
    if (!info->is_object()) {
        if (res != SRS_HTTP_RESPONSE_OK) {
            ret = ERROR_HTTP_DATA_INVALID;
            srs_error("invalid response number %s. ret=%d", res.c_str(), ret);
            return ret;
        }
        return ret;
    }

    // response standard object, format in json: {"code": 0, "data": ""}
    auto res_info = info->to_object();
    auto res_code = res_info->ensure_property_integer("code");
    if (res_code == NULL) {
        ret = ERROR_RESPONSE_CODE;
        srs_error("invalid response without code, ret=%d", ret);
        return ret;
    }

    if ((res_code->to_integer()) != ERROR_SUCCESS) {
        ret = ERROR_RESPONSE_CODE;
        srs_error("error response code=%d. ret=%d", res_code->to_integer(), ret);
        return ret;
    }

    res_code = res_info->ensure_property_string("data");
    if (res_code == nullptr) {
        res.clear();
    } else {
        res.assign(res_code->to_str());
    }
    return ret;
}

#endif
